﻿using System;
using System.Collections.Generic;
using System.Configuration;
using System.Linq;
using System.Messaging;
using System.Reactive.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Derbster.Peer.Receiver
{
    public class MessageReceiver : IMessageReceiver
    {
        private static MessageQueue _messageQueue;
        private readonly CancellationTokenSource _cancellationTokenSource;

        public bool IsEmpty
        {
            get
            {
                return _messageQueue.GetAllMessages().Length == 0;
            }
        }

        public MessageReceiver()
        {
            if (_messageQueue == null)
            {
                var multicastAddress = ConfigurationManager.AppSettings["multicastAddress"];
                if (String.IsNullOrWhiteSpace(multicastAddress)) multicastAddress = "224.42.42.1:8191";

                string queuePath = ".\\private$\\Derbster_" + PeerContext.PeerId;
                if (!MessageQueue.Exists(String.Format(queuePath)))
                {
                    _messageQueue = MessageQueue.Create(queuePath, false);
                }
                else
                {
                    _messageQueue = new MessageQueue(queuePath);
                }

                _messageQueue.MulticastAddress = multicastAddress;
                _messageQueue.UseJournalQueue = true;
                _messageQueue.SetFormatter<PeerMessage>();
            }
            
            _cancellationTokenSource = new CancellationTokenSource();
        }


        public void Run()
        {
            Observable.FromAsync(_messageQueue.RecieveAsync)
                .Repeat()
                .Subscribe(ProcessQueueMessage, _cancellationTokenSource.Token);
        }

        public void Stop()
        {
            _cancellationTokenSource.Cancel();
        }

        public void Dispose()
        {
            _cancellationTokenSource.Cancel();
            _messageQueue.Dispose();
        }

        protected void ProcessQueueMessage(Message message)
        {
            var peerMessage = (PeerMessage) message.Body;

            var receivedVectorClock = peerMessage.VectorClock;
            PeerContext.CurrentVectorClock.Increment();
            PeerContext.CurrentVectorClock.MergeReceived(receivedVectorClock);
        }
    }
}
